#1. 读取文件的数据test.txt
from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import *
spark = SparkSession.builder.appName('homework2').getOrCreate()
schema = StructType([
    StructField('班级', IntegerType(), True),
    StructField('姓名', StringType(), True),
    StructField('年龄', IntegerType(), True),
    StructField('性别', StringType(), True),
    StructField('科目', StringType(), True),
    StructField('成绩', IntegerType(), True)
])
df = spark.read.csv('data.csv', header=True, schema=schema)

#2. 一共有多少个小于20岁的人参加考试？
df.filter(df['年龄'] < 20).count()

# 3. 一共有多少个等于20岁的人参加考试？
df.filter(df['年龄'] == 20).select('姓名').distinct().count()

# 4. 一共有多少个大于20岁的人参加考试？
df.filter(df['年龄'] > 20).select('姓名').distinct().count()

# 5. 一共有多个男生参加考试？
df.filter(df['性别'] == '男').select('姓名').distinct().count()

# 6. 一共有多少个女生参加考试？
df.filter(df['性别'] == '女').select('姓名').distinct().count()

# 7. 12班有多少人参加考试？
df.filter(df['班级'] == '12').select('姓名').distinct().count()

# 8. 13班有多少人参加考试？
df.filter(df['班级'] == '13').select('姓名').distinct().count()

# 9. 语文科目的平均成绩是多少？
df.filter(df['科目'] == 'chinese').select(avg(col('成绩'))).show()

# 10. 数学科目的平均成绩是多少？
df.filter(df['科目'] == 'math').select(avg(col('成绩'))).show()

# 11. 英语科目的平均成绩是多少？
df.filter(df['科目'] == 'english').select(avg(col('成绩'))).show()

# 12. 每个人平均成绩是多少？
df.groupBy(['班级', '姓名']).agg({'成绩':'avg'}).show()

# 13. 12班平均成绩是多少？
df.filter(df['班级'] == '12').select(avg('成绩')).show()

# 14. 12班男生平均总成绩是多少？
tmp_df = df.filter(df['班级'] == '12').filter(df['性别'] == '男').groupBy(['班级', '姓名']).agg({'成绩':'sum'})
tmp_df.select(avg(col('sum(成绩)'))).show()

# 15. 12班女生平均总成绩是多少？
tmp_df = df.filter(df['班级'] == '12').filter(df['性别'] == '女').groupBy(['班级', '姓名']).agg({'成绩':'sum'})
tmp_df.select(avg(col('sum(成绩)'))).show()

# 16. 13班平均成绩是多少？
df.filter(df['班级'] == '13').select(avg('成绩')).show()

# 17. 13班男生平均总成绩是多少？
tmp_df = df.filter(df['班级'] == '13').filter(df['性别'] == '男').groupBy(['班级', '姓名']).agg({'成绩':'sum'})
tmp_df.select(avg(col('sum(成绩)'))).show()

# 18. 13班女生平均总成绩是多少？
tmp_df = df.filter(df['班级'] == '13').filter(df['性别'] == '女').groupBy(['班级', '姓名']).agg({'成绩':'sum'})
tmp_df.select(avg(col('sum(成绩)'))).show()

# 19. 全校语文成绩最高分是多少？
df.filter(df['科目'] == 'chinese').select(max(col('成绩'))).show()

# 20. 12班语文成绩最低分是多少？
df.filter(df['科目'] == 'chinese').select(min(col('成绩'))).show()

# 21. 13班数学最高成绩是多少？
df.filter(df['科目'] == 'math').select(max(col('成绩'))).show()

# 22. 总成绩大于150分的12班的女生有几个？
tmp_df = df.filter(df['班级'] == 12).filter(df['性别'] == '女').groupBy(['班级', '姓名']).agg({'成绩':'sum'})
tmp_df.filter(tmp_df['sum(成绩)'] > 150).count()

# 23. 总成绩大于150分，且数学大于等于70，且年龄大于等于19岁的学生的平均成绩是多少？
age_df = df.filter(df['年龄'] >= 19)
sum_df = age_df.groupBy(['班级', '姓名']).agg({'成绩':'sum'})
avg_df = age_df.groupBy(['班级', '姓名']).agg({'成绩':'avg'})
mat_df = age_df.filter(df['科目'] == 'math').filter(df['成绩'] >= 70)
tmp_df = mat_df.join(sum_df, '姓名').join(avg_df, '姓名')
tmp_df.filter(tmp_df['sum(成绩)'] >= 150).select('姓名', 'avg(成绩)').show()